package com.atguigu.flink.chapter10.query;


import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/*
分组开窗聚合
 */
public class QueryWindowDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //在ddl中定义时间属性
        tEnv.executeSql("create table sensor(" +
                "id string," +
                "ts bigint," +
                "vc int," +
                "et as to_timestamp_ltz(ts,3)," +
                "watermark for et as et - interval '1' second " +
                ") with(" +
                "  'connector' = 'filesystem', " +
                "  'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\sensor.txt', " +
                "  'format' = 'csv' " +
                ")");

        //滚动窗口
        tEnv.sqlQuery("select " +
                "id," +
                "tumble(et,interval '5' second) stt," +
                "tumble(et,interval '5' second) edt," +
                "sum(vc) vc_sum " +
                "from sensor " +
                "group by id,tumble(et,interval '5' second)")
                .execute()
                .print();


        //滑动窗口
        tEnv.sqlQuery("select " +
                        "id," +
                        "hop_start(et,interval '2' second,interval '5' second) stt," +
                        "hop_end(et,interval '2' second,interval '5' second) edt," +
                        "sum(vc) vc_sum " +
                        "from sensor " +
                        "group by id,hop(et,interval '2' second,interval '5' second)")
                .execute()
                .print();

        //会话窗口
        tEnv.sqlQuery("select " +
                        "id," +
                        "session_start(et,interval '2' second) stt," +
                        "session_end(et,interval '2' second) edt," +
                        "sum(vc) vc_sum " +
                        "from sensor " +
                        "group by id,session(et,interval '2' second)")
                .execute()
                .print();
    }
}


//public class QueryWindowDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//
//        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
//
//        //在ddl中定义时间属性
//        tEnv.executeSql("create table sensor(" +
//                "id string," +
//                "ts bigint," +
//                "vc int," +
//                "et as to_timestamp_ltz(ts,3)," +
//                "watermark for et as et - interval '1' second" +
//                ") with (" +
//                "  'connector' = 'filesystem', " +
//                "  'path' = 'C:\\IDEA\\code\\LearnDemo\\flink\\input\\sensor.txt', " +
//                "  'format' = 'csv' " +
//                ")");
//
//        //滑动窗口
//        tEnv.sqlQuery("select " +
//                "id," +
//                "hop_strat(et,interval '2' second,interval '5' second) stt," +
//                "hop_end(et,interval '2' second,interval '5' second) edt," +
//                "sum(vc) vc_sum" +
//                "from sensor " +
//                "group by id,hop(et,interval '2' second,interval '5' second)")
//                .execute()
//                .print();
//
//        //滚动窗口
//        tEnv.sqlQuery("select " +
//                        "id," +
//                        "tumble_strat(et,interval '5' second) stt," +
//                        "tumble_end(et,interval '5' second) edt," +
//                        "sum(vc) vc_sum" +
//                        "from sensor " +
//                        "group by id,tumble(et,interval '5' second)")
//                .execute()
//                .print();
//
//
//        //会话窗口
//        tEnv.sqlQuery("select " +
//                        "id," +
//                        "session_strat(et,interval '5' second) stt," +
//                        "session_end(et,interval '5' second) edt," +
//                        "sum(vc) vc_sum" +
//                        "from sensor " +
//                        "group by id,session(et,interval '5' second)")
//                .execute()
//                .print();
//    }
//}